#include "Avalon/Network/TcpServerSocket.hpp"
#include <boost/bind.hpp>
#include <boost/lexical_cast.hpp>
#include "Avalon/IO/NotConnectedException.hpp"
#include "Avalon/Network/SocketException.hpp"
#include "Avalon/Network/SocketType.hpp"
#include "Avalon/Network/TcpSocketConnection.hpp"
#include "Avalon/Network/TcpSocketReader.hpp"
#include "Avalon/Network/TcpSocketWriter.hpp"
#include "Avalon/Network/SocketThreadPool.hpp"

using namespace Avalon;
using namespace Avalon::IO;
using namespace Avalon::Network;
using namespace Avalon::Network::Detail;
using namespace Avalon::Threading;
using namespace boost;
using namespace boost::asio;
using namespace boost::posix_time;
using namespace boost::signals2;
using namespace std;
using boost::asio::ip::tcp;

TcpServerSocket::TcpServerSocket(const IpAddress& address)
    : m_address(address),
      m_isConnected(false),
      m_ioService(SocketThreadPool::GetInstance().GetService()) {}

TcpServerSocket::~TcpServerSocket() {
  Close();
}

void TcpServerSocket::Start() {
  boost::lock_guard<recursive_mutex> lock(m_mutex);
  if(m_isConnected) {
    return;
  }
  tcp::resolver resolver(m_ioService);
  tcp::resolver::query query(m_address.first,
    lexical_cast<string>(m_address.second));
  boost::system::error_code error;
  tcp::resolver::iterator endpointIterator = resolver.resolve(query, error);
  if(error != 0) {
    BOOST_THROW_EXCEPTION(SocketException(error.value(),
      error.message().c_str()));
  }
  m_acceptor = new tcp::acceptor(m_ioService, *endpointIterator);
  m_isConnected = true;
}

Async<Channel*>::Ptr TcpServerSocket::Accept() {
  boost::lock_guard<recursive_mutex> lock(m_mutex);
  if(!m_isConnected) {
    BOOST_THROW_EXCEPTION(NotConnectedException());
  }
  AsyncOwner<Channel*>::Ptr acceptResult = AsyncOwner<Channel*>::Create();
  m_acceptResult = acceptResult;
  tcp::acceptor& r_acceptor = *static_cast<tcp::acceptor*>(m_acceptor);
  boost::shared_ptr<SocketType> socket(new SocketType(m_ioService));
  r_acceptor.async_accept(socket->m_socket,
    bind(&TcpServerSocket::HandleAccept, this, socket, placeholders::error));
  return acceptResult;
}

bool TcpServerSocket::IsConnected() {
  boost::lock_guard<recursive_mutex> lock(m_mutex);
  return m_isConnected;
}

void TcpServerSocket::Close() {
  boost::unique_lock<recursive_mutex> lock(m_mutex);
  if(!m_isConnected) {
    return;
  }
  tcp::acceptor& r_acceptor = *static_cast<tcp::acceptor*>(m_acceptor);
  r_acceptor.close();
  delete static_cast<tcp::acceptor*>(m_acceptor);
  m_isConnected = false;
  if(m_acceptResult != NULL) {
    m_acceptCondition.wait(lock);
  }
  m_closedSignal();
}

connection TcpServerSocket::ConnectClosedSignal(
    const ClosedSignal::slot_type& slot) {
  boost::lock_guard<recursive_mutex> lock(m_mutex);
  return m_closedSignal.connect(slot);
}

connection TcpServerSocket::ConnectClosedSignal(
    const ClosedSignal::slot_type& slot, bool* isConnected) {
  boost::lock_guard<recursive_mutex> lock(m_mutex);
  *isConnected = m_isConnected;
  return m_closedSignal.connect(slot);
}

void TcpServerSocket::HandleAccept(const boost::shared_ptr<SocketType>& socket,
    const boost::system::error_code& error) {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  AsyncOwner<Channel*>::Ptr acceptResult = m_acceptResult;
  m_acceptResult.reset();
  tcp::acceptor& r_acceptor = *static_cast<tcp::acceptor*>(m_acceptor);
  if(!error) {
    boost::shared_ptr<TcpSocketReader> reader(new TcpSocketReader(socket));
    boost::shared_ptr<TcpSocketWriter> writer(new TcpSocketWriter(socket));
    boost::shared_ptr<TcpSocketConnection> connection(new TcpSocketConnection(
      true, IpAddress(socket->m_socket.remote_endpoint().address().to_string(),
      socket->m_socket.remote_endpoint().port()), socket, reader, writer));
    TcpClientSocketChannel* channel = new TcpClientSocketChannel(connection,
      reader, writer);
    acceptResult->SetResult(channel);
  } else if(error == asio::error::operation_aborted && m_isConnected) {
    m_acceptResult = acceptResult;
    r_acceptor.async_accept(socket->m_socket, bind(
      &TcpServerSocket::HandleAccept, this, socket, placeholders::error));
  } else {
    acceptResult->SetException(SocketException(error.value(),
      error.message().c_str()));
  }
  m_acceptCondition.notify_all();
}
